package com.bw.yk03;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.phoenix.jdbc.PhoenixDriver;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;


//{"database":"gmall2021","data":{"area_code":"540000","name":"西藏","region_id":"6","iso_3166_2":"CN-XZ","id":34,"iso_code":"CN-54"},"type":"insert","table":"base_province"}
//{"area_code":"540000","name":"西藏","region_id":"6","iso_3166_2":"CN-XZ","id":34,"iso_code":"CN-54"}
/*
create table dim_user_info(id varchar primary key, line varchar);
create table dim_sku_info(id varchar primary key, line varchar);
create table dim_base_province(id varchar primary key, line varchar);
 */
public class FlinkCDC2SinkHbase extends RichSinkFunction<String> {
    private String tableName;
    private Connection conn = null;
    PreparedStatement ps = null;
    public FlinkCDC2SinkHbase() {
    }

    public FlinkCDC2SinkHbase(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        conn = DriverManager.getConnection("jdbc:phoenix:hadoop-single:2181", "", "");
        ps = conn.prepareStatement("upsert into " + tableName + " values(?,?)");
        conn.setAutoCommit(true);
    }

    @Override
    public void close() throws Exception {
        ps.close();
        conn.close();
    }

    @Override
    public void invoke(String value, Context context) throws Exception {
        JSONObject jo = JSON.parseObject(value);
        JSONObject data = jo.getJSONObject("data");
        String id = data.getString("id");
        String line = data.toString();
        ps.setString(1,id);
        ps.setString(2,line);
        ps.execute();

    }
}
